package com.stm.bi.etl.sync;

import com.stm.bi.core.util.DirPathUtil;
import com.stm.bi.jdbc.DataSourceEnum;
import com.stm.bi.jdbc.DataSourceSwitch;
import com.stm.bi.repository.AppOrganizationRepository;
import com.stm.bi.repository.DorisLoadRepository;
import com.stm.bi.repository.VerToBetaDorisLoadRepository;
import com.stm.framework.HadoopProvider;
import com.stm.framework.core.util.ClassUtil;
import com.stm.framework.core.util.SpringContextUtil;
import com.stm.framework.core.util.StringUtil;
import com.stm.framework.dbcp.DataSourceConfig;
import com.stm.framework.dbcp.JdbcService;
import com.stm.framework.dto.Response;
import com.stm.framework.redisson.annotation.SyncLock;
import com.stm.ucenter.dto.vo.org.OrganizationVO;
import jakarta.annotation.Resource;
import org.apache.hadoop.fs.FileStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.sql.DataSource;
import java.io.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @Author chenyingjun
 * @Desc
 * @Date 2024/12/19 10:23
 */
@Component
public class AdsDataSynchronizer {
    private static final Logger log = LoggerFactory.getLogger(AdsDataSynchronizer.class);
    private static final String HADOOP_PATH = "/user/hive/warehouse/ads.db";
    private static final String SELECT_DORIS_TABLE_SQL = "select table_name from information_schema.tables where table_schema=?";

    @Resource
    private JdbcService jdbcService;

    @Resource
    private HadoopProvider hadoopProvider;
    @Resource
    private DirPathUtil dirPathUtil;

    private final static String COLUMN_SEPARATOR = "|";

    @Resource
    private AppOrganizationRepository appOrganizationRepository;
    
    @Resource
    private DorisLoadRepository dorisLoadRepository;

    @Resource
    private VerToBetaDorisLoadRepository verToBetaDorisLoadRepository;

    private String[] schemas = {"dmp_ads","dmp_paisi_ads"};



    /**
     * 同步广告数据到Doris表
     * 该方法使用了@SyncLock注解来确保在同一时间只有一个线程可以执行同步操作，避免数据冲突
     * 注解参数key指定了同步锁的键，这里使用固定的键名'ads_data_synchronizer'来控制整个同步过程
     */
    @SyncLock(key="'ads_data_synchronizer'",leaseTime = 1000*60*20)
    public Response synchronize() {
        // 获取当前类的实例，用于调用其他方法
        AdsDataSynchronizer _this = SpringContextUtil.getBean(getClass());
        for(String schema : schemas){
            // 获取所有Doris表名
            List<String> dorisTables = _this.getDorisTables(schema);

            // 获取Hadoop文件系统中指定路径下的所有文件状态
            List<FileStatus> tableFileStatusList = hadoopProvider.listFiles(HADOOP_PATH);

            // 获取所有组织信息
            List<OrganizationVO> organizations = appOrganizationRepository.selectOrgList();

            // 遍历每个文件状态和每个组织，以确定需要同步的Doris表
            for(FileStatus tableFileStatus : tableFileStatusList){
                for(OrganizationVO org : organizations){
                    // 获取Hive表名
                    String hiveTableName = tableFileStatus.getPath().getName();
                    // 通过组织编号和Hive表名构造Doris表名
                    String dorisTableName = dorisTables.stream().filter(x->hiveTableName.equalsIgnoreCase(x+"_"+org.getNo())).findFirst().orElse(null);
                    // 如果找到了匹配的Doris表名，则进行数据同步
                    if(dorisTableName!=null){
                        // 日志记录数据同步开始
                        log.info("数据同步开始,doris表={},hive表{}",dorisTableName,hiveTableName);
                        // 获取Hive表的列名
                        List<String> hiveTableColumns = _this.getHiveTableColumns(hiveTableName);
                        // 获取数据文件的相对路径
                        String tablePath = getRelativePath(tableFileStatus);
                        // 获取指定路径下的所有数据文件状态
                        List<FileStatus> dataFileStatusList = hadoopProvider.listFiles(tablePath);
                        // 遍历每个数据文件状态，执行数据同步操作
                        for(FileStatus dataFileStatus : dataFileStatusList){
                            synchronize(dataFileStatus,hiveTableColumns,schema,dorisTableName);
                        }
                        // 日志记录数据同步结束
                        log.info("数据同步结束,doris表={},hive表{}",dorisTableName,hiveTableName);
                    }
                }
            }
        }


        return Response.success();
    }

    /**
     * 同步组织信息和文件数据到目标数据库表
     *
     * @param fileStatus 文件状态对象，包含文件或目录的元数据信息
     * @param hiveColumns 列名列表，用于映射文件数据到数据库表的列
     * @param dorisTableName 目标数据库表名，用于插入数据
     */
    private void synchronize(FileStatus fileStatus,List<String> hiveColumns,String schema,String dorisTableName) {
        // 判断当前处理的是目录还是文件
        if(fileStatus.isDirectory()){
            // 如果是目录，获取相对路径并列出目录下的所有文件或子目录
            String dirPath = getRelativePath(fileStatus);
            List<FileStatus> fileStatusList = hadoopProvider.listFiles(dirPath);
            // 递归调用synchronize方法处理目录下的每个文件或子目录
            for(FileStatus status : fileStatusList){
                synchronize(status,hiveColumns,schema,dorisTableName);
            }
        }else{
            // 如果是文件，从HDFS中获取文件并开始处理
            try {
                File hdfsFile = getHdfsFile(fileStatus);
                String columns = hiveColumns.stream().collect(Collectors.joining(","));
                dorisLoadRepository.putLoad(schema, dorisTableName, hdfsFile,columns,COLUMN_SEPARATOR);

                //VER TO BETA
                verToBetaDorisLoadRepository.putLoad(schema+"_beta",dorisTableName,hdfsFile,columns,COLUMN_SEPARATOR);

                hadoopProvider.deleteFile(getRelativePath(fileStatus));
                hdfsFile.delete();
                log.error("文件同步成功,dorisTableName={}",dorisTableName);
            } catch (IOException e) {
                log.error("文件同步失败,fileName={}",fileStatus.getPath().getName(),e);
            }
        }
    }

    /**
     * 根据FileStatus获取HDFS文件
     * 该方法将HDFS上的文件下载到本地临时路径
     *
     * @param fileStatus HDFS文件的状态信息，包含文件的路径等
     * @return 返回下载到本地的文件对象
     */
    private File getHdfsFile(FileStatus fileStatus) throws IOException {
        // 获取文件名
        String fileName= fileStatus.getPath().getName();
        // 记录日志，文件下载开始
        log.info("文件下载开始,fileName={}",fileName);
        // 记录开始时间
        Long startTime = System.currentTimeMillis();
        // 拼接临时文件路径
        String tempPath = dirPathUtil.getBasePath()+"/"+ fileName+".csv";
        // 创建临时文件对象
        File tempFile = new File(tempPath);
        // 创建文件输出流
        OutputStream outputStream = new FileOutputStream(tempFile);
        // 获取数据文件相对路径
        String dataFilePath = getRelativePath(fileStatus);
        // 调用HadoopProvider的文件下载方法，将HDFS文件下载到本地
        hadoopProvider.downloadFile(dataFilePath,outputStream);
        // 刷新输出流
        outputStream.flush();
        // 关闭输出流
        outputStream.close();
        // 记录日志，文件下载结束
        log.info("文件下载结束,fileName={},cost={}ms",fileName,System.currentTimeMillis()-startTime);
        // 返回临时文件对象
        return tempFile;
    }

    /**
     * 获取文件的相对路径
     * 该方法用于从FileStatus对象中提取出路径，并将其转换为相对于HADOOP_PATH的路径
     * 主要用于在处理文件时，去除文件路径中的冗余部分，保留从HADOOP_PATH开始的路径信息
     *
     * @param fileStatus 文件的元数据信息，包含文件的路径等信息
     * @return 返回相对于HADOOP_PATH的路径字符串
     */
    private String getRelativePath(FileStatus fileStatus) {
        String path = fileStatus.getPath().toString();
        int idx = path.lastIndexOf(HADOOP_PATH);
        String relativePath = path.substring(idx);
        return relativePath;
    }

    /**
     * 获取Doris数据库中的所有表名
     * 该方法通过切换到ADS数据源，并执行SQL查询来获取Doris数据库中的所有表名
     * 主要用于需要展示或操作Doris表时，获取其表名列表
     *
     * @return 返回Doris数据库中所有表的表名列表
     */
    //DataSourceEnum

    public List<String> getDorisTables(String schema) {
        List<Map> rowDatas = jdbcService.selectList(SELECT_DORIS_TABLE_SQL, new Object[]{schema});
        return rowDatas.stream().map(x->(String)x.get("table_name")).collect(Collectors.toList());
    }

    /**
     * 获取Hive表中的所有列名
     * 该方法通过切换到DMP数据源，并执行SQL查询来获取指定Hive表中的所有列名
     * 主要用于需要展示或操作Hive表结构时，获取其列名列表
     *
     * @param tableName 指定的Hive表名
     * @return 返回指定Hive表的所有列名列表
     */
    @DataSourceSwitch(DataSourceEnum.DMP)
    public List<String> getHiveTableColumns(String tableName) {
        List<Map> rowDatas = jdbcService.selectList("DESCRIBE ads."+tableName, new Object[]{});
        List<String> columns = new ArrayList<>();
        for(Map rowData : rowDatas){
            String columnName = (String)rowData.get("col_name");
            if(StringUtil.isEmpty(columnName)){
                continue;
            }
            if(columnName.startsWith("#")){
                continue;
            }
            if(columnName.equalsIgnoreCase("ms")){
                continue;
            }
            if(columns.contains(columnName)){
                continue;
            }
            columns.add(columnName);
        }
        return columns;
    }

}
